/*
 * Copyright (C) 2020 Agrui
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <stdlib.h>
#include <string.h>
#include <MessageQueue.h>
#include <Message.h>
#include <MessagePool.h>

MessageQueue* newMessageQueue(boolean quitAllowed) {
    MessageQueue* queue = (MessageQueue*)malloc(sizeof(MessageQueue));
    if (!queue) {
        return NULL;
    }

    memset(queue, 0, sizeof(MessageQueue));

    queue->mLock = newLock();
    if (!queue->mLock) {
        free(queue);
        return NULL;
    }

    queue->mSig = newSignal();
    if (!queue->mSig) {
        deleteLock(queue->mLock);
        free(queue);
        return NULL;
    }

    queue->mQuitAllowed = quitAllowed;
    messagePoolInit();
    return queue;
}

void deleteQueue(MessageQueue* queue) {
    if (queue) {
        if (queue->mSig) {
            deleteSignal(queue->mSig);
            queue->mSig = NULL;
        }

        if (queue->mLock) {
            deleteLock(queue->mLock);
            queue->mLock = NULL;
        }

        free(queue);
        messagePoolDeInit();
    }

}

boolean removeMessages(MessageQueue* queue, Handler* handler, int what, void* object) {
    Message* p;

    if (!queue || queue->mLock || !handler) {
        return FALSE;
    }

    p = queue->mMessages;

    // Remove all messages at front.
    while (p != NULL && p->target == handler && p->what == what
           && (object == NULL || p->obj == object)) {
        Message* n = p->next;
        queue->mMessages = n;
        messageRecycleUnchecked(p);
        p = n;
    }

    // Remove all messages after front.
    while (p != NULL) {
        Message* n = p->next;
        if (n != NULL) {
            if (n->target == handler && n->what == what
                && (object == NULL || n->obj == object)) {
                Message* nn = n->next;
                messageRecycleUnchecked(n);
                p->next = nn;
                continue;
            }
        }
        p = n;
    }
}

void removeAllMessagesLocked(MessageQueue* queue) {
    Message* p;
    if (!queue) {
        return;
    }
    p = queue->mMessages;
    while (p != NULL) {
        Message* n = p->next;
        messageRecycleUnchecked(p);
        p = n;
    }
    queue->mMessages = NULL;
}


Message* next(MessageQueue* queue) {
    if (!queue) {
        return NULL;
    }

    for (;;) {

        lock(queue->mLock);
        Message* prevMsg = NULL;
        Message* msg = queue->mMessages;
        if (msg != NULL && msg->target == NULL) {
            // Stalled by a barrier.  Find the next asynchronous message in the queue.
            do {
                prevMsg = msg;
                msg = msg->next;
            } while (msg != NULL);
        }
        if (msg != NULL) {
            // Got a message.
            //mBlocked = false;
            if (prevMsg != NULL) {
                prevMsg->next = msg->next;
            } else {
                queue->mMessages = msg->next;
            }
            msg->next = NULL;
            //msg->markInUse();
            unlock(queue->mLock);
            return msg;
        }

        if (queue->mQuitting) {
            unlock(queue->mLock);
            return NULL;
        }

        unlock(queue->mLock);

        if (msg == NULL) {
            // No more messages.
            //nextPollTimeoutMillis = -1;
            signalWait(queue->mSig, 0);
        }
    }

    return NULL;
}

void quit(MessageQueue* queue) {
    if (queue) {
        if (!queue->mQuitAllowed) {
            return;
        }

        lock(queue->mLock);
        if (queue->mQuitting) {
            unlock(queue->mLock);
            return;
        }

        queue->mQuitting = TRUE;
        removeAllMessagesLocked(queue);
        unlock(queue->mLock);

        setSignal(queue->mSig);
    }
}

boolean enqueueMessage(MessageQueue* queue, Message* msg) {
    Message* p = queue->mMessages;

    if (!queue || !queue->mLock || !msg || !msg->target) {
        return FALSE;
    }

    lock(queue->mLock);

    //boolean needWake;
    if (p == NULL) {
        // New head, wake up the event queue if blocked.
        msg->next = p;
        queue->mMessages = msg;
    } else {
        // Inserted within the middle of the queue.  Usually we don't have to wake
        // up the event queue unless there is a barrier at the head of the queue
        // and the message is the earliest asynchronous message in the queue.
        Message* prev;
        for (;;) {
            prev = p;
            p = p->next;
            if (p == NULL) {
                break;
            }
        }
        msg->next = p; // invariant: p == prev.next
        prev->next = msg;
    }

    unlock(queue->mLock);
    if (!queue->mSig) {
        return FALSE;
    }
    setSignal(queue->mSig);
    return TRUE;
}


